<?php

declare(strict_types=1);

namespace App\Cola\Amqp\Consumer;

use App\Cola\Logic\ColaImportBill;
use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Message\Type;
use Hyperf\Amqp\Result;
use PhpAmqpLib\Message\AMQPMessage;
use Hyperf\DbConnection\Db;
use App\Cola\Model\ColaBill;
use App\Cola\Model\ColaTradeCate;
use App\Common\Excel;
use Hyperf\Di\Annotation\Inject;

#[Consumer(exchange: 'bill_import_exchange', routingKey: 'bill_import_routing_key', queue: 'bill_import_queue', name: "BillImportProducer", nums: 1)]
class BillImportConsumer extends ConsumerMessage
{

    public function isEnable(): bool
    {
        return parent::isEnable();
    }

    public function consumeMessage($data, AMQPMessage $message): Result
    {

        try {
            // 处理消息
            $taskId = $data['task_id'];
            $userId = $data['user_id'];
            $paySource = $data['source'];
            $file = $data['file'];

            // 记录任务开始状态
            cache()->set("bill_import_task:{$taskId}", json_encode(['status' => 0, 'msg' => '处理中']));

            // 处理账单导入逻辑
            $server = new ColaImportBill;
            $result = $server->example($taskId,$userId,$paySource,$file);

            // 更新任务状态
            if ($result) {
                cache()->set("bill_import_task:{$taskId}", json_encode(['status' => 1, 'msg' => '导入成功']));
            } else {
                cache()->set("bill_import_task:{$taskId}", json_encode(['status' => 2, 'msg' => '导入失败']));
            }
            return Result::ACK;
        } catch (\Exception $e) {

            logger()->info('[账单同步]失败', exception_to_array($e));

            // 记录异常日志
            return Result::NACK;
        }
    }

    public function getType(): Type
    {
        return Type::DIRECT;
    }


}
